package consumer;

import base.TopicModule;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;


/**
 * @author : zGame
 * @version V1.0
 * @Project: rokectmqprac
 * @Package consumer
 * @Description: TODO
 * @date Date : 2022年01月11日 13:05
 */
public class NormalConsumer extends ConsumerUtil {
    
    @Override
    public void propertySet() {
        try {
            consumer.setNamesrvAddr("localhost:9876");
            //一个GroupName第一次消费时的位置
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            //consumer.setConsumeThreadMin(20);
            //consumer.setConsumeThreadMax(20);
            //要消费的topic，可使用tag进行简单过滤
            consumer.subscribe(getTopic(getCurrentServerId()), "*");
            //一次最大消费的条数
            consumer.setConsumeMessageBatchMaxSize(100);
            //消费模式，广播或者集群，默认集群。
            consumer.setMessageModel(MessageModel.CLUSTERING);
            //在同一jvm中 需要启动两个同一GroupName的情况需要这个参数不一样。
            consumer.setInstanceName(this.getClass().getSimpleName());
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
    
    @Override
    public TopicModule getTopicModule() {
        return TopicModule.GAME;
    }
}
